package com.coolw.dynamic.thread.pool.context;

import com.coolw.common.exception.ExceptionUtil;
import com.coolw.dynamic.thread.pool.constant.CommonStat;
import com.coolw.dynamic.thread.pool.domain.dto.ThreadPoolStatDTO;
import com.coolw.dynamic.thread.pool.entity.ThreadPoolConfigEntity;
import com.coolw.dynamic.thread.pool.enums.KeepAliveTimeUnitEnum;
import com.coolw.dynamic.thread.pool.enums.QueueTypeEnum;
import com.coolw.dynamic.thread.pool.enums.RejectedExecutionHandlerEnum;
import com.coolw.dynamic.thread.pool.enums.StatStatusEnum;
import lombok.extern.slf4j.Slf4j;
import org.dromara.hutool.core.collection.CollUtil;
import org.dromara.hutool.core.date.DateFormatPool;
import org.dromara.hutool.core.date.DateUtil;
import org.dromara.hutool.core.text.StrUtil;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 动态线程池管理器
 *
 * @author coolw
 * @date 2024/5/16 10:55
 */
@Slf4j
public class DynamicThreadPoolManager {

    /**
     * 线程池执行器(key:线程池编码,value:线程池执行器)
     */
    private static final Map<String, ThreadPoolExecutor> EXECUTOR_MAP = new ConcurrentHashMap<>();

    /**
     * 线程池统计成功计数
     */
    private final AtomicLong successCount = new AtomicLong(0);

    private static volatile DynamicThreadPoolManager dynamicThreadPoolManager;

    private DynamicThreadPoolManager() {
    }

    public static DynamicThreadPoolManager getInstance() {
        if (dynamicThreadPoolManager == null) {
            synchronized (DynamicThreadPoolManager.class) {
                if (dynamicThreadPoolManager == null) {
                    dynamicThreadPoolManager = new DynamicThreadPoolManager();
                }
            }
        }
        return dynamicThreadPoolManager;
    }

    public static List<String> getKeys() {
        return new ArrayList<>(EXECUTOR_MAP.keySet());
    }

    public static ThreadPoolExecutor getExecutor(String poolCode) {
        return EXECUTOR_MAP.get(poolCode);
    }

    public synchronized void createThreadPool(List<ThreadPoolProperties> properties) {
        if (CollUtil.isEmpty(properties)) {
            return;
        }

        properties.forEach(p -> {
            ThreadPoolExecutor executor = newThreadPoolExecutor(p);
            EXECUTOR_MAP.put(p.getPoolCode(), executor);
        });
        log.info("线程池已全部创建,poolCodes:{}", getKeys());
    }

    public synchronized void closeThreadPool() {
        if (CollUtil.isEmpty(EXECUTOR_MAP)) {
            return;
        }

        EXECUTOR_MAP.forEach((poolCode, executor) -> executor.shutdown());
        log.info("线程池已全部关闭,poolCodes:{}", getKeys());
    }

    public synchronized boolean createThreadPool(ThreadPoolProperties properties) {
        String poolCode = properties.getPoolCode();
        ThreadPoolExecutor executor = getExecutor(poolCode);
        if (Objects.nonNull(executor)) {
            log.info("线程池已存在,poolCode:{}", poolCode);
            return true;
        }

        executor = newThreadPoolExecutor(properties);
        EXECUTOR_MAP.put(poolCode, executor);
        log.info("线程池已创建,poolCode:{}", poolCode);
        return true;
    }

    public synchronized boolean modifyThreadPool(ThreadPoolProperties properties) {
        String poolCode = properties.getPoolCode();
        ThreadPoolExecutor executor = getExecutor(poolCode);
        if (Objects.isNull(executor)) {
            log.info("线程池配置不存在,poolCode:{}", poolCode);
            return false;
        }

        // 核心线程数
        int corePoolSize = properties.getCorePoolSize();
        if (corePoolSize != executor.getCorePoolSize()) {
            executor.setCorePoolSize(corePoolSize);
        }

        // 最大线程数
        int maxPoolSize = properties.getMaxPoolSize();
        if (maxPoolSize != executor.getMaximumPoolSize()) {
            executor.setMaximumPoolSize(maxPoolSize);
        }

        // 线程存活时间及单位
        long keepAliveTime = properties.getKeepAliveTime();
        TimeUnit timeUnit = KeepAliveTimeUnitEnum.getTimeUnit(properties.getKeepAliveTimeUnit());
        if (keepAliveTime != executor.getKeepAliveTime(timeUnit)) {
            executor.setKeepAliveTime(keepAliveTime, timeUnit);
        }

        // 拒绝策略
        RejectedExecutionHandler rejectedExecutionHandler = getRejectedExecutionHandler(properties.getRejectStrategy());
        if (rejectedExecutionHandler.getClass() != executor.getRejectedExecutionHandler().getClass()) {
            executor.setRejectedExecutionHandler(rejectedExecutionHandler);
        }

        log.info("线程池已修改,poolCode:{},properties:{}", poolCode, properties);
        return true;
    }

    private ThreadPoolExecutor newThreadPoolExecutor(ThreadPoolProperties properties) {
        String poolCode = properties.getPoolCode();
        if (EXECUTOR_MAP.containsKey(poolCode)) {
            return EXECUTOR_MAP.get(poolCode);
        }

        TimeUnit timeUnit = KeepAliveTimeUnitEnum.getTimeUnit(properties.getKeepAliveTimeUnit());
        RejectedExecutionHandler handler = getRejectedExecutionHandler(properties.getRejectStrategy());
        BlockingQueue<Runnable> blockQueue = getBlockQueue(properties.getQueueType(), properties.getQueueSize());
        ThreadFactory threadFactory = new DefaultThreadFactory(poolCode);
        return new ThreadPoolExecutor(properties.getCorePoolSize()
                , properties.getMaxPoolSize()
                , properties.getKeepAliveTime()
                , timeUnit
                , blockQueue
                , threadFactory
                , handler);
    }

    /**
     * 线程池统计
     */
    public synchronized ThreadPoolStatDTO monitor(List<ThreadPoolConfigEntity> configList) {
        if (CollUtil.isEmpty(configList)) {
            return null;
        }

        Map<String, ThreadPoolConfigEntity> configMap = configList.stream()
                .collect(Collectors.toMap(ThreadPoolConfigEntity::getPoolCode, c -> c));

        ThreadPoolStatDTO stat = new ThreadPoolStatDTO();
        stat.setStartTime(now());
        List<ThreadPoolStatDTO.StatInfo> statInfoList = new ArrayList<>(configList.size());
        EXECUTOR_MAP.forEach((key, value) -> {
            ThreadPoolConfigEntity config = configMap.get(key);
            ThreadPoolStatDTO.StatInfo statInfo;
            try {
                statInfo = buildStatInfo(config, value);
                statInfo.setStatStatus(StatStatusEnum.SUCCESS.getCode());
            } catch (Exception e) {
                log.error("线程池统计异常,config:{}", config, e);
                statInfo = new ThreadPoolStatDTO.StatInfo();
                statInfo.setStatStatus(StatStatusEnum.FAIL.getCode());
            }
            statInfoList.add(statInfo);
        });
        stat.setEndTime(now());
        stat.setSuccessCount(successCount.incrementAndGet());
        stat.setStatInfoList(statInfoList);
        return stat;
    }

    private ThreadPoolStatDTO.StatInfo buildStatInfo(ThreadPoolConfigEntity config, ThreadPoolExecutor executor) {
        // 运行中的队列使用大小
        int useQueueSize = executor.getQueue().size();
        // 配置中的队列大小
        int queueSize = config.getQueueSize();

        ThreadPoolStatDTO.StatInfo statInfo = new ThreadPoolStatDTO.StatInfo();
        statInfo.setPoolNo(config.getPoolNo());
        statInfo.setPoolName(config.getPoolName());
        statInfo.setPoolCode(config.getPoolCode());
        statInfo.setQueueType(config.getQueueType());
        statInfo.setQueueSize(queueSize);
        // 当前队列使用大小
        statInfo.setUseQueueSize(useQueueSize);
        // 当前队列使用占比
        statInfo.setUseQueueRate(calcUseQueueRata(queueSize, useQueueSize));
        // 核心线程数
        statInfo.setCorePoolSize(executor.getCorePoolSize());
        // 最大线程数
        statInfo.setMaxPoolSize(executor.getMaximumPoolSize());
        // 线程存活时间
        statInfo.setKeepAliveTime(executor.getKeepAliveTime(KeepAliveTimeUnitEnum.getTimeUnit(config.getKeepAliveTimeUnit())));
        // 线程存活时间单位
        statInfo.setKeepAliveTimeUnit(config.getKeepAliveTimeUnit());
        // 拒绝策略
        statInfo.setRejectStrategy(covertRejectStrategy(executor.getRejectedExecutionHandler()));
        // 线程池中活动线程数量
        statInfo.setActiveCount(executor.getActiveCount());
        // 已执行完成的任务量
        statInfo.setCompletedTaskCount(executor.getCompletedTaskCount());
        // 执行的任务量(包括正在执行的任务)
        statInfo.setTaskCount(executor.getTaskCount());
        // 线程池中曾有过的最大线程量
        statInfo.setLargestPoolSize(executor.getLargestPoolSize());
        return statInfo;
    }

    /**
     * 计算队列使用占比
     */
    private String calcUseQueueRata(int totalQueueSize, int useQueueSize) {
        BigDecimal bigTotalQueueSize = BigDecimal.valueOf(totalQueueSize);
        BigDecimal bigUseQueueSize = BigDecimal.valueOf(useQueueSize);
        return bigUseQueueSize
                .multiply(CommonStat.HUNDRED)
                .divide(bigTotalQueueSize, 2, RoundingMode.HALF_UP)
                .toString();
    }

    /**
     * 获取阻塞队列
     */
    private BlockingQueue<Runnable> getBlockQueue(String queueType, int queueSize) {
        if (StrUtil.equals(QueueTypeEnum.ARRAY_BLOCKING_QUEUE.getCode(), queueType)) {
            return new ArrayBlockingQueue<>(queueSize);
        }

        if (StrUtil.equals(QueueTypeEnum.LINKED_BLOCKING_QUEUE.getCode(), queueType)) {
            return new LinkedBlockingQueue<>(queueSize);
        }

        throw ExceptionUtil.createBusinessException(StrUtil.format("[{}]队列类型未定义", queueType));
    }

    /**
     * 获取拒绝策略
     */
    private RejectedExecutionHandler getRejectedExecutionHandler(String rejectStrategy) {
        if (StrUtil.equals(rejectStrategy, RejectedExecutionHandlerEnum.DISCARD_OLDEST_POLICY.getCode())) {
            return new ThreadPoolExecutor.DiscardOldestPolicy();
        }

        if (StrUtil.equals(rejectStrategy, RejectedExecutionHandlerEnum.CALLER_RUNS_POLICY.getCode())) {
            return new ThreadPoolExecutor.CallerRunsPolicy();
        }

        if (StrUtil.equals(rejectStrategy, RejectedExecutionHandlerEnum.ABORT_POLICY.getCode())) {
            return new ThreadPoolExecutor.AbortPolicy();
        }

        if (StrUtil.equals(rejectStrategy, RejectedExecutionHandlerEnum.DISCARD_POLICY.getCode())) {
            return new ThreadPoolExecutor.DiscardPolicy();
        }

        return new ThreadPoolExecutor.AbortPolicy();
    }

    private String covertRejectStrategy(RejectedExecutionHandler handler) {
        if (handler instanceof ThreadPoolExecutor.DiscardOldestPolicy) {
            return RejectedExecutionHandlerEnum.DISCARD_OLDEST_POLICY.getCode();
        }

        if (handler instanceof ThreadPoolExecutor.CallerRunsPolicy) {
            return RejectedExecutionHandlerEnum.CALLER_RUNS_POLICY.getCode();
        }

        if (handler instanceof ThreadPoolExecutor.AbortPolicy) {
            return RejectedExecutionHandlerEnum.ABORT_POLICY.getCode();
        }

        if (handler instanceof ThreadPoolExecutor.DiscardPolicy) {
            return RejectedExecutionHandlerEnum.DISCARD_POLICY.getCode();
        }

        // 支持自定义拒绝策略......待完成
        throw ExceptionUtil.createBusinessException("拒绝策略未定义");
    }

    private static String now() {
        return DateUtil.format(LocalDateTime.now(), DateFormatPool.NORM_DATETIME_PATTERN);
    }
}
